package hbase.mapReduce;

import hbase.common.MyPut;
import hbase.tools.Constants;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class StudentInfoApp {

    public static class SIMapper extends TableMapper<Text, NullWritable>{

        Text outKey = new Text();

        @Override
        protected void map(ImmutableBytesWritable key, Result value, Mapper<ImmutableBytesWritable, Result, Text, NullWritable>.Context context) throws IOException, InterruptedException {

            String name = new String(value.getValue(Bytes.toBytes("baseInfo"),Bytes.toBytes("name")));
            outKey.set(name);
            context.write(outKey,NullWritable.get());
        }
    }

    public static class SIReducer extends TableReducer<Text, NullWritable,ImmutableBytesWritable>{

        ImmutableBytesWritable outKey = new ImmutableBytesWritable();

        int count = 0;

        @Override
        protected void reduce(Text key, Iterable<NullWritable> values, Reducer<Text, NullWritable, ImmutableBytesWritable, Mutation>.Context context) throws IOException, InterruptedException {

            String rowKey = ++count + "";

            outKey.set(Bytes.toBytes(rowKey));

            MyPut put = new MyPut(rowKey);
            put.addColumn("result","name",key.toString());

            context.write(outKey,put);

        }
    }

    public static void main(String[] args) throws Exception {

        Configuration conf = HBaseConfiguration.create();
        conf.set(Constants.ZK_QUORUM,Constants.ZK_QUORUM_NODES);
        conf.set(Constants.ZK_CLIENT_PORT,Constants.ZK_PORT);

        Job job = Job.getInstance(conf);

        Scan scan = new Scan();
        TableMapReduceUtil.initTableMapperJob(
                "s1",
                scan,
                SIMapper.class,
                Text.class,
                NullWritable.class,
                job
        );

        TableMapReduceUtil.initTableReducerJob(
                "s2",
                SIReducer.class,
                job
        );
        job.waitForCompletion(true);
    }

}
